package com.atguigu.upp.app;

import com.atguigu.upp.beans.TagInfo;
import com.atguigu.upp.beans.TagValueTypeConstant;
import com.atguigu.upp.service.CKService;
import com.atguigu.upp.service.MysqlService;
import com.atguigu.upp.util.UPPUtil;
import org.apache.ibatis.session.SqlSessionFactory;
import org.apache.spark.sql.SparkSession;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

/**
 * Created by Smexy on 2023/3/27
 */
public class GenerateUserGroupApp
{
    public static void main(String[] args) throws IOException {

        //接收画像平台传入的参数
        String taskId = args[0];
        String doDate = args[1];

       /* String taskId = "3";
        String doDate = "2022-06-09";*/

        SqlSessionFactory mysqlSSF = UPPUtil.createSqlSessionFactory("mysqlconfig.xml");
        SqlSessionFactory ckSSF = UPPUtil.createSqlSessionFactory("ckconfig.xml");
        MysqlService mysqlService = new MysqlService(mysqlSSF.openSession(true));
        CKService ckService = new CKService(ckSSF.openSession(true));

        // 当前平台统一使用Spark-submit提交应用，当前必须是一个SparkApp( 有SparkContext，或SparkSession )
        SparkSession sparkSession = UPPUtil.getSparkSession("GenerateUserGroupApp");

        //查询今天宽边中计算了哪些标签，根据标签的tag_value_type对标签进行分类。分为四类
        List<TagInfo> tagInfos = mysqlService.queryTags();

        List<TagInfo> stringTags = new ArrayList<>();
        List<TagInfo> bigintTags = new ArrayList<>();
        List<TagInfo> decimalTags = new ArrayList<>();
        List<TagInfo> dateTags = new ArrayList<>();

        for (TagInfo tagInfo : tagInfos) {
            switch (tagInfo.getTagValueType()) {
                case TagValueTypeConstant.TAG_VALUE_TYPE_LONG:
                    bigintTags.add(tagInfo);
                    break;
                case TagValueTypeConstant.TAG_VALUE_TYPE_DECIMAL:
                    decimalTags.add(tagInfo);
                    break;
                case TagValueTypeConstant.TAG_VALUE_TYPE_STRING:
                    stringTags.add(tagInfo);
                    break;
                case TagValueTypeConstant.TAG_VALUE_TYPE_DATE:
                    dateTags.add(tagInfo);
                    break;
            }
        }

        //以类型为组，初步分群
        generateUserGroup(stringTags, ckService, doDate, "user_tag_value_string");
        generateUserGroup(bigintTags, ckService, doDate, "user_tag_value_long");
        generateUserGroup(decimalTags, ckService, doDate, "user_tag_value_decimal");
        generateUserGroup(dateTags, ckService, doDate, "user_tag_value_date");


    }

    private static void generateUserGroup(List<TagInfo> tags, CKService ckService, String doDate, String table) {

        if (tags.size() > 0) {

            //先删除今天已经写入的数据
            ckService.deleteBitmap(table, doDate);

            String db = UPPUtil.getValue("updbname");
            String source = UPPUtil.getValue("upwideprefix") + doDate.replace("-", "_");
            String tagStr = tags.stream()
                                .map(t -> String.format(" ('%s',%s) ", t.getTagCode().toLowerCase(), t.getTagCode().toLowerCase()))
                                .collect(Collectors.joining(","));

            //写入
            ckService.generateUserGroup(table, doDate, tagStr, db, source);

        }

    }
}
